package com.sunway.resource2es.service;


import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.Refresh;
import co.elastic.clients.elasticsearch.core.BulkRequest;
import co.elastic.clients.elasticsearch.core.BulkResponse;
import co.elastic.clients.elasticsearch.indices.CreateIndexRequest;
import co.elastic.clients.elasticsearch.indices.CreateIndexResponse;
import co.elastic.clients.elasticsearch.indices.IndexSettings;
import co.elastic.clients.transport.endpoints.BooleanResponse;
import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import com.sunway.exception.built.BusinessException;
import com.sunway.resource2es.util.EsMappingUtil;
import com.sunway.resource2es.util.MyStrUtil;
import com.sunway.result.utils.UUIDUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.core.io.ClassPathResource;
import org.springframework.stereotype.Service;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.util.*;

/**
 * 数据ES索引操作Service
 *
 * @author lxw
 * @version V1.0
 * @date 2024-7-18
 */
//@Slf4j
@Service
public class EsProcessServiceImpl {

    private static final Logger log = LoggerFactory.getLogger(EsProcessServiceImpl.class);

    private final JdbcSqlServiceImpl jdbcSqlService;
    private final ElasticsearchClient client;

    public EsProcessServiceImpl(JdbcSqlServiceImpl jdbcSqlService
            , ElasticsearchClient client) {
        this.jdbcSqlService = jdbcSqlService;
        this.client = client;
    }

    /**
     * 项目中，已存在的mapping文件名称，带文件格式.json，如ruc_report.json
     *
     * @param mappingFileName mapping文件名称
     * @param mappingStr      mapping 规则内容
     * @return com.alibaba.fastjson2.JSONObject
     * @author lxw
     * @date 2025/3/20 14:04
     **/
    public JSONObject loadMapping(String mappingFileName, String mappingStr) {
        try {
            if (StringUtils.isNotBlank(mappingFileName)) {
                ClassPathResource basicMappingResource = new ClassPathResource("mapping/" + mappingFileName);

                // 从 ClassPathResource 获取 InputStream
                InputStream inputStream = basicMappingResource.getInputStream();
                // 将 InputStream 转换为 String
                mappingStr = new String(inputStream.readAllBytes(), StandardCharsets.UTF_8);
            }
            if (StringUtils.isBlank(mappingStr)) {
                throw new BusinessException("读取mapping规则失败！");
            }
            // 将 JSON 字符串转换为 JSONObject
            return JSONObject.parseObject(mappingStr);
        } catch (IOException e) {
            log.error("读取mapping规则失败！{}，{}", mappingFileName, e.getMessage());
            throw new BusinessException("读取mapping规则失败！" + e.getMessage());
        }
    }

    /**
     * 生成ES索引结构
     **/
    public boolean createEsIndex(String indexName, String indexAlias, String mapping) {

        try {
            if (StringUtils.equalsIgnoreCase(indexName, indexAlias)) {
                throw new BusinessException("索引名称与索引别名不能相同！");
            }
            BooleanResponse exists = client.indices().exists(e -> e.index(indexName));
            if (exists.value()) {
                boolean acknowledged = client.indices().delete(d -> d.index(indexName)).acknowledged();
                log.info("生成ES索引结构，索引已存在，所以删除索引！：{}", acknowledged);
            }
            // 将 JSON 字符串转换为 InputStream
            InputStream inputStream = new ByteArrayInputStream(mapping.getBytes());
            // setting 创建
            IndexSettings.Builder settingBuilder = EsMappingUtil.initSetting(1, 0);
            CreateIndexRequest createIndexRequest = new CreateIndexRequest.Builder()
                    .index(indexName)
                    .aliases(indexAlias, a -> a)
                    .settings(settingBuilder.build())
                    .mappings(m -> m.withJson(inputStream))
                    .build();
            CreateIndexResponse indexResponse = client.indices().create(createIndexRequest);
            boolean created = Boolean.TRUE.equals(indexResponse.acknowledged());
            log.info("生成ES索引结构,索引创建响应: {}", created);
            // 创建索引
            return created;
        } catch (IOException e) {
            log.error("生成ES索引结构，创建索引失败，e--->:{}", e.getMessage());
            throw new BusinessException("创建索引失败！");
        }
    }

    public List<String> initEsData(String dbName, String tableName, String condition, String mappingFileName, String mappingStr, String indexName, String indexAlias) {
        // 读取索引规则
        JSONObject mappingJson = loadMapping(mappingFileName, mappingStr);

        Object properties = mappingJson.get("properties");
        if (properties == null) {
            throw new BusinessException("索引规则为空！");
        }
        String jsonString = createJsonString(properties);
        if (!createEsIndex(indexName, indexAlias, jsonString)) {
            throw new BusinessException("创建索引失败！");
        }
        return importEsData(dbName, tableName, condition, indexName, mappingJson);
    }

    public List<String> importEsData(String dbName, String tableName, String condition, String indexName, JSONObject mappingJson) {

        Object properties = mappingJson.get("properties");
        if (properties == null) {
            throw new BusinessException("索引规则为空！");
        }
        String onlyId = Optional.ofNullable(mappingJson.get("request_id_field")).map(Object::toString).orElse("");
        // ES索引数据值为字符串数组，且允许重复的字段
        List<String> arrayFieldsAllowRepeat = processListString(mappingJson.get("array_field_allow_repeat"));
        // ES索引数据值为字符串数组，且不允许重复的字段
        List<String> arrayFieldsNotRepeat = processListString(mappingJson.get("array_field_not_repeat"));
        // ES索引数据值为数组对象的字段
        List<String> jsonArrayField = processListString(mappingJson.get("json_array_field"));
        // ES索引数据值为对象的字段
        List<String> jsonObjectField = processListString(mappingJson.get("json_object_field"));

        JSONObject from = JSONObject.from(properties);
        Set<String> keys = from.keySet();
        List<String> fieldList = new ArrayList<>(keys);

        // 处理响应，查找失败的操作
        List<String> failedIds = new ArrayList<>();
        Long count = jdbcSqlService.getTableCount(dbName, tableName, condition);
        int cardinality = 500;
        int total = (int) Math.ceil(count / (double) cardinality);
        String lastId = "0";
        for (int i = 0; i < total; i++) {
            // 查询数据

            List<Map<String, Object>> list = jdbcSqlService.fetchData(dbName, tableName, condition, lastId, cardinality);
            if (list.isEmpty()) continue;
            lastId = list.get(list.size() - 1).get("id").toString();
            // 加入ES
            BulkResponse bulkResponse = processBulkRequest(indexName, list, onlyId, fieldList, arrayFieldsAllowRepeat, arrayFieldsNotRepeat, jsonArrayField, jsonObjectField);
            if (bulkResponse != null && bulkResponse.errors()) {
                // 处理错误数据
                handleBulkErrors(bulkResponse, failedIds);
            }
            count -= cardinality;
            log.info("{}数据表，第{}次批量操作，剩余{}条数据", tableName, (i + 1), (count > 0 ? count : 0));
        }
        return failedIds;
    }


    /**
     * 创建 mapping json字符串
     *
     * @param properties 字段配置
     * @return java.lang.String
     * @author lxw
     * @date 2024/12/9 9:16
     **/
    private String createJsonString(Object properties) {
        Map<String, Object> mm = new HashMap<>();
        mm.put("properties", properties);
        return JSONObject.toJSONString(mm);
    }

    /**
     * 数据加入ES
     *
     * @param indexName              索引名称
     * @param list                   原数据
     * @param onlyId                 原数据中唯一值字段
     * @param fieldList              ES索引需要的字段
     * @param arrayFieldsAllowRepeat ES索引数据值为字符串数组，且允许值重复
     * @param arrayFieldsNotRepeat   ES索引数据值为字符串数组，且不允许值重复
     * @param jsonArrayField         ES索引数据值为数组对象的字段，对象数组
     * @param jsonObjectField        ES索引数据值为对象的字段，对象
     * @return co.elastic.clients.elasticsearch.core.BulkResponse
     * @author lxw
     * @date 2024/12/9 9:18
     **/
    private BulkResponse processBulkRequest(String indexName, List<Map<String, Object>> list, String onlyId, List<String> fieldList,
                                            List<String> arrayFieldsAllowRepeat, List<String> arrayFieldsNotRepeat
            , List<String> jsonArrayField, List<String> jsonObjectField) {
        BulkRequest.Builder bulkRequest = new BulkRequest.Builder();
        for (Map<String, Object> map : list) {
            JSONObject jsonObject = processArrayFields(map, fieldList, arrayFieldsAllowRepeat, arrayFieldsNotRepeat, jsonArrayField, jsonObjectField);
            String id = Optional.ofNullable(map.get(onlyId)).map(Object::toString).orElse(UUIDUtils.getInstance().generateShortUuid());
            bulkRequest.operations(op -> op.index(idx -> idx.index(indexName).id(id).document(jsonObject)));
        }
        bulkRequest.refresh(Refresh.True);
        try {
            return client.bulk(bulkRequest.build());
        } catch (IOException e) {
            log.error("{}数据表批量创建索引，ES错误: {}", indexName, e.getMessage());
            return null;
        }
    }

    /**
     * 数据转换为ES索引需要的数据
     *
     * @param map                    原数据
     * @param fieldList              ES索引字段集合
     * @param arrayFieldsAllowRepeat ES索引数据值为字符串数组，且允许值重复
     * @param arrayFieldsNotRepeat   ES索引数据值为字符串数组，且不允许值重复
     * @param jsonArrayField         ES索引数据值为数组对象的字段，对象数组
     * @param jsonObjectField        ES索引数据值为对象的字段，对象
     * @return java.util.Map<java.lang.String, java.lang.Object>
     * @author lxw
     * @date 2024/12/9 9:20
     **/
    private JSONObject processArrayFields(Map<String, Object> map, List<String> fieldList, List<String> arrayFieldsAllowRepeat, List<String> arrayFieldsNotRepeat
            , List<String> jsonArrayField, List<String> jsonObjectField) {

        arrayFieldsAllowRepeat.forEach(field -> map.put(field, processArrayString(map.get(field), false)));
        arrayFieldsNotRepeat.forEach(field -> map.put(field, processArrayString(map.get(field), true)));
        jsonArrayField.forEach(field -> map.put(field, processArrayObject(map.get(field))));
        jsonObjectField.forEach(field -> map.put(field, processObject(map.get(field))));
        JSONObject jsonObject = new JSONObject();
        fieldList.forEach(key -> jsonObject.put(key, map.get(key)));
        return jsonObject;
    }

    /**
     * 字符串 || 分割为数组
     *
     * @param fieldValue 字段值
     * @param distinct   是否允许值重复
     * @return java.util.List<java.lang.String>
     * @author lxw
     * @date 2024/12/9 9:21
     **/
    private List<String> processArrayString(Object fieldValue, boolean distinct) {
        if (fieldValue == null) return null;
        return distinct ? MyStrUtil.str2ListNotRepeat(fieldValue.toString()) : MyStrUtil.str2ListAllowRepeat(fieldValue.toString());
    }

    /**
     * Object 转 JSONObject
     *
     * @param fieldValue 字段值
     * @return com.alibaba.fastjson2.JSONObject
     * @author lxw
     * @date 2024/12/9 10:43
     **/
    private JSONObject processObject(Object fieldValue) {
        if (fieldValue != null) {
            try {
                return JSONObject.parseObject(fieldValue.toString());
            } catch (Exception e) {
                log.error("Object 转 JSONObject 失败！ {}", e.getMessage());
            }
        }
        return null;
    }

    /**
     * Object 转 JSONArray
     *
     * @param fieldValue 字段值
     * @return com.alibaba.fastjson2.JSONArray
     * @author lxw
     * @date 2024/12/9 10:43
     **/
    private JSONArray processArrayObject(Object fieldValue) {
        if (fieldValue != null) {
            try {
                return JSONArray.parseArray(fieldValue.toString());
            } catch (Exception e) {
                log.error("Object 转 JSONArray 失败！ {}", e.getMessage());
            }
        }
        return null;
    }

    /**
     * Object 转 List<String>
     *
     * @param fieldValue 字段值
     * @return java.util.List<java.lang.String>
     * @author lxw
     * @date 2024/12/9 10:45
     **/
    private static List<String> processListString(Object fieldValue) {
        return (fieldValue == null) ? Collections.emptyList() : JSONArray.parseArray(fieldValue.toString(), String.class);
    }


    /**
     * 索引结果处理
     *
     * @param bulkResponse ES批次结果
     * @param failedIds    错误数据ID集合
     * @author lxw
     * @date 2024/12/9 9:25
     **/
    private void handleBulkErrors(BulkResponse bulkResponse, List<String> failedIds) {
        bulkResponse.items().stream()
                .filter(item -> item.error() != null)
                .forEach(item -> failedIds.add(item.id()));
    }

}
